package cn.itcast.czxy

import base.BaseModel
import org.apache.spark.sql.{DataFrame, Dataset, Row}

object ConstellationTags extends BaseModel {
  //导入隐式转化
  import spark.implicits._
  import org.apache.spark.sql.functions._

  //重写程序名称
  override def sparkName(): String = "constellation"

  //重写四五级标签的 id
  override def getId: Int = 125

  //主方法
  def main(args: Array[String]): Unit = {
    exec()
  }


  //重写抽象方法  计算、匹配标签
  override def getTags(fiveRow: Dataset[Row], hbaseDF: DataFrame): DataFrame = {
    val fiveDF: DataFrame = fiveRow.map(row => {
      val id: String = row.getAs("id").toString
      val rule: String = row.getAs("rule").toString
      //1.20-2.18
      val arr: Array[String] = rule.split("#")
      var start = ""
      var end = ""
      if (arr != null && arr.length == 2) {
        start = arr(0)
        end = arr(1)
      }
      (id, start, end)
    }).toDF("id", "start", "end")

    fiveDF.createOrReplaceTempView("fiveDF")
    hbaseDF.createOrReplaceTempView("hbaseDF")

//    spark.sql(
//      """
//        |SELECT id,CONCAT('2020-',start)as start,CONCAT('2020-',end) as end
//        |FROM fiveDF
//        |""".stripMargin)
//
//    spark.sql(
//      """
//        |SELECT id,CONCAT('2020-',SUBSTRING(birthday,6,10)) as birthday
//        |FROM hbaseDF
//        |""".stripMargin)

    spark.sql(
      """
        |SELECT tmp_h.id as userId,
        |CASE WHEN tmp_f.id IS NULL THEN '137' ELSE tmp_f.id END as tagsId
        |FROM
        |(SELECT id,CONCAT('2020-',START) AS `start`,CONCAT('2020-',END) AS `end`
        |FROM fiveDF)tmp_f
        |RIGHT JOIN
        |(SELECT id,CONCAT('2020-',SUBSTRING(birthday,6,10)) AS birthday
        |FROM hbaseDF)tmp_h
        |ON tmp_h.birthday BETWEEN tmp_f.`start` AND tmp_f.`end`
        |""".stripMargin)

  }
}
